package cn.com.sparkle.paxos.paxosinstance.paxossender;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.Context;
import cn.com.sparkle.paxos.model.Id;
import cn.com.sparkle.paxos.model.SuccessTransportConfig;
import cn.com.sparkle.paxos.model.Value;
import cn.com.sparkle.paxos.net.client.system.SystemNetNode;
import cn.com.sparkle.paxos.net.client.system.callback.PaxosPrepareCallBack;
import cn.com.sparkle.paxos.net.client.system.callback.PaxosVoteCallBack;
import cn.com.sparkle.paxos.net.client.system.callback.PrepareCallBack;
import cn.com.sparkle.paxos.net.client.system.callback.VoteCallBack;
import cn.com.sparkle.paxos.paxosinstance.PaxosInstance;
import cn.com.sparkle.paxos.route.RouteManage.LinkedNodeList;

public class HalfCircleInstancePaxosMessageSender extends AbstractInstancePaxosMessageSender {

	private final static Logger logger = Logger.getLogger(HalfCircleInstancePaxosMessageSender.class);
	private LinkedNodeList voteLink;
	private LinkedList<SuccessTransportConfig> notifyShip = new LinkedList<SuccessTransportConfig>();

	private LinkedList<SuccessTransportConfig> successList;
	private SystemNetNode successMessageHeadNode;
	private SuccessTransportConfig successMessageHeadTransportConfig;

	@SuppressWarnings("unchecked")
	public HalfCircleInstancePaxosMessageSender(Context context, LinkedNodeList voteLink, int quorum) {
		super(quorum);
		this.voteLink = voteLink;
		//calc notify ship
		int scalePerNode = (int) Math.ceil((double) voteLink.getRemainedNodes().size() / (voteLink.getList().size() - 1));//ignore one node joined vote.
		LinkedList<String> notifyList = new LinkedList<String>();
		notifyList.addAll(voteLink.getRemainedNodes());
		Iterator<String> iter = voteLink.getList().descendingIterator();
		String lastNode = null;
		while (iter.hasNext()) {
			String node = iter.next();

			if (lastNode != null) {
				List<String> list = new LinkedList<String>();
				for (int i = 0; i < scalePerNode; ++i) {
					if (notifyList.size() != 0) {
						String address = notifyList.remove();
						list.add(address);
					}
				}
				SuccessTransportConfig config = new SuccessTransportConfig(lastNode, list, false);
				notifyShip.addFirst(config);
			}

			lastNode = node;
		}
		//calc success head node
		if (voteLink.getHeadNode().equals(context.getConfiguration().getSelfAddress())) {
			if (notifyShip.size() > 0) {
				successMessageHeadNode = (SystemNetNode) context.getcState().getSenators().getValidActiveNodes()
						.get(notifyShip.getFirst().getTargetAddress());
				successList = (LinkedList<SuccessTransportConfig>) notifyShip.clone();
				successMessageHeadTransportConfig = successList.removeFirst();
			}
		} else {
			successMessageHeadNode = voteLink.getHeadNode();
		}
		if (context.getConfiguration().isDebugLog()) {
			StringBuffer stringBuffer = new StringBuffer();
			stringBuffer.append("voteNodes:[ ");
			stringBuffer.append(voteLink.getList().getFirst() + " ");
			for (SuccessTransportConfig transport : notifyShip) {
				stringBuffer.append(transport.getTargetAddress()).append("(notifyShip:");
				for (String notifyAddress : transport.getTargetNotifyAddress()) {
					stringBuffer.append(notifyAddress).append(",");
				}
				stringBuffer.append(") ");
			}
			stringBuffer.append("]");
			logger.debug(stringBuffer.toString());
		}

	}

	@Override
	public void sendSuccessRequest(long instanceId, Id id, Value value) {
		//find first 		
		try {
			if (successMessageHeadNode == voteLink.getHeadNode()) { 
				//the head is not self
				successMessageHeadNode.sendInstanceSuccessMessage(instanceId, id, null, null, this.notifyShip);
			} else if (successMessageHeadNode != null) {
				//the head is self, so send the next head
				successMessageHeadNode.sendInstanceSuccessMessage(instanceId, id,
						successMessageHeadTransportConfig.isNotifyTransValue() ? value : null, successMessageHeadTransportConfig.getTargetNotifyAddress(), successList);
			}
		} catch (InterruptedException e) {
			//may be lost the headNode, cancel the process ,and wait other node to study the successful instance actively.
		}
	}

	@Override
	public void sendPrepareRequest(PaxosInstance paxosInstance, long instanceId, Id id) {
		PrepareCallBack callback = new PaxosPrepareCallBack(1, 1, paxosInstance);
		voteLink.getHeadNode().sendInstancePrepareRequest(id, instanceId, voteLink.getNoHeadList(), callback);
	}

	@Override
	public void sendVoteRequest(PaxosInstance paxosInstance, long instanceId, Id id, Value value) {
		VoteCallBack callback = new PaxosVoteCallBack(1, 1, paxosInstance);
		voteLink.getHeadNode().sendInstanceVoteRequest(instanceId, id, value, voteLink.getNoHeadList(), callback);
	}
}
